package com.yujiahao.bigdata.rdd.action

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}


/**TODO 测试行动算子
    1、这几个没啥滑头，明白含义直接用就行
        first()：返回数据集中的第一个。take(N)：拿到前N条数据。takeOrdered(N)：先排序在拿到前N条数据。
    2、下面使用的时候有讲究。
      这个aggregate第一个列表表示初始值，存在柯里化，第二个列表分别表示分区内，和分区的计算规则。
      重点是初始化的值在分区间和分区内都会用到
      aggregate(这个是写初始值的，分区内或者分区外计算都要使用的）（第一个参数表示分区内的计算规则，第二个参数就是分区间的计算规则）
    3、aggregate的特殊情况，分区内和分区间的计算规则相同的时候，可以使用fold
      fold(这个是写初始值的，分区内或者分区外计算都要使用的)(计算规则)
*/
object Spark01_Action_First {
  def main(args: Array[String]): Unit = {
    //TODO 1、获得Spark环境的连接z
    // 参数解读：setAppName("当前执行程序的名字").setMaster("当前的环境")
    // 分布式环境的基本架构为 主 - 从
    val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 3、业务逻辑
    val rdd: RDD[Int] = sc.makeRDD(List(1, 4, 3, 2), 2)

    //TODO 行动算子 first ：返回数据集中的第一个
    val i: Int = rdd.first()
    println("测试行动算子 first ：返回数据集中的第一个：" + i)

    //TODO 行动算子 take ：返回数据集中的前几个，返回的是个数组
    val array: Array[Int] = rdd.take(3)
    println("测试行动算子 take ：返回数据集中的前几个，返回的是个数组：" + array.mkString(" "))


    //TODO 行动算子 takeOrdered :获取当前处理数据集中的前几条数据，先排序在取前N条。

    val ints: Array[Int] = rdd.takeOrdered(3)
    //先排序在取值
    println("测试行动算子 takeOrdered：先排序在取前N条:" + ints.mkString(" "))


    //TODO 行动算子aggregate：聚合算子
    // aggregateByKey(初始值)(分区内的计算规则，分区间的计算规则)
    //aggregate:存在函数柯里化，与两个参数列表
    //第一个参数列表中，只有一个参数表示计算初始值,这个初始值，在计算内会使用，在分区间计算时也会使用
    //第二个参数列表中，有两个参数，第一个参数表示分区内计算规则，第二个规则表示分区间计算规则
    //TODO 如果分区内和分区间计算规则相同的时候可以简化使用fold

    val i1: Int = rdd.aggregate(10)(_ + _, _ + _)
    val i2: Int = rdd.fold(10)(_ + _)
    println("测试行动算子aggregate：聚合算子:" + i1)
    println("测试行动算子fold：聚合算子:" + i2)


    //TODO 2、关闭Spark环境的连接
    sc.stop()

  }

}
